package com.cm;

import com.alibaba.fastjson.JSON;
import com.cm.pojo.Goods;
import org.apache.http.HttpHost;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;


public class EsHigherDemo {

    private RestHighLevelClient client;

    /**
     * 创建连接
     */
    @Before
    public void connect() {
        client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("192.168.206.99", 9200, "http")
                ));
    }

    /**
     * 关闭连接
     */
    @After
    public void close() throws IOException {
        client.close();
    }

    /**
     * 创建索引库 以及 映射规则
     */
    @Test
    public void testCreateIndex() throws IOException {
        CreateIndexRequest request = new CreateIndexRequest("goods");
        request.source("{\n" +
                "  \"settings\": {\n" +
                "    \"analysis\": {\n" +
                "      \"analyzer\": {\n" +
                "        \"my_pinyin\": {\n" +
                "          \"tokenizer\": \"ik_smart\",\n" +
                "          \"filter\": [\n" +
                "            \"py\"\n" +
                "          ]\n" +
                "        }\n" +
                "      },\n" +
                "      \"filter\": {\n" +
                "        \"py\": {\n" +
                "          \"type\": \"pinyin\",\n" +
                "          \"keep_full_pinyin\": false,\n" +
                "          \"keep_joined_full_pinyin\": true,\n" +
                "          \"keep_original\": true,\n" +
                "          \"limit_first_letter_length\": 16,\n" +
                "          \"remove_duplicated_term\": true\n" +
                "        }\n" +
                "      }\n" +
                "    }\n" +
                "  },\n" +
                "  \"mappings\": {\n" +
                "    \"properties\": {\n" +
                "      \"id\": {\n" +
                "        \"type\": \"keyword\"\n" +
                "      },\n" +
                "      \"name\": {\n" +
                "        \"type\": \"completion\",\n" +
                "        \"analyzer\": \"my_pinyin\",\n" +
                "        \"search_analyzer\": \"ik_smart\"\n" +
                "      },\n" +
                "      \"title\":{\n" +
                "        \"type\": \"text\",\n" +
                "        \"analyzer\": \"my_pinyin\",\n" +
                "        \"search_analyzer\": \"ik_smart\"\n" +
                "      },\n" +
                "      \"price\":{\n" +
                "        \"type\": \"long\"\n" +
                "      }\n" +
                "    }\n" +
                "  }\n" +
                "}", XContentType.JSON);
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
        System.out.println("isAcknowledged：" + response.isAcknowledged());
    }


    /**
     * 批量新增
     */
    @Test
    public void testAddBulk() throws IOException {
        // 1.准备文档数据
        List<Goods> list = new ArrayList<>();
        list.add(new Goods(1L, "红米9", "红米9手机 数码", 1499L));
        list.add(new Goods(2L, "三星 Galaxy A90", "三星 Galaxy A90 手机 数码 疾速5G 骁龙855", 3099L));
        list.add(new Goods(3L, "Sony WH-1000XM3", "Sony WH-1000XM3 降噪耳机 数码", 2299L));
        list.add(new Goods(4L, "松下剃须刀", "松下电动剃须刀高转速磁悬浮马达", 599L));
        BulkRequest request = new BulkRequest();
        for (Goods goods : list) {
            request.add(new IndexRequest("goods")
                    .id(goods.getId().toString())
                    .source(JSON.toJSONString(goods), XContentType.JSON));
        }
        BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
        System.out.println("bulkResponse.status() = " + bulkResponse.status());
    }


    /**
     * 删除索引库
     */
    @Test
    public void testDeleteIndex() throws IOException {
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest("goods");
        AcknowledgedResponse response = client.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
        System.out.println("response = " + response.isAcknowledged());
    }


    /**
     * 文档新增
     */
    @Test
    public void testAddDoc() throws IOException {
        IndexRequest request = new IndexRequest("goods");
        request.id("114");
        request.source("{\n" +
                "  \"id\":\"114\",\n" +
                "  \"name\":\"小米手机\",\n" +
                "  \"title\":\"小米手机，超好用的手机\",\n" +
                "  \"price\":2899\n" +
                "}", XContentType.JSON);
        IndexResponse response = client.index(request, RequestOptions.DEFAULT);
        System.out.println("response.status() = " + response.status());
    }


    /**
     * 文档的查询
     */
    @Test
    public void testFindDoc() throws IOException {
        GetRequest goodsRequest = new GetRequest("goods", "114");
        GetResponse response = client.get(goodsRequest, RequestOptions.DEFAULT);
        String source = response.getSourceAsString();
        Goods goods = JSON.parseObject(source, Goods.class);
        System.out.println("goods = " + goods);
    }


    /**
     * 文档的修改
     */
    @Test
    public void testUpdateDoc() throws IOException {
        UpdateRequest updateRequest = new UpdateRequest("goods", "114");
        updateRequest.doc("price", 899);
        UpdateResponse response = client.update(updateRequest, RequestOptions.DEFAULT);
        System.out.println("response.status() = " + response.status());
    }


    /**
     * 文档的删除
     */
    @Test
    public void testDeleteDoc() throws IOException {
        DeleteRequest deleteRequest = new DeleteRequest("goods", "114");
        DeleteResponse response = client.delete(deleteRequest, RequestOptions.DEFAULT);
        System.out.println("response.status() = " + response.status());
    }

    /**
     * 查询所有
     */
    @Test
    public void testFindAll() throws IOException {
        SearchRequest searchRequest = new SearchRequest("goods");
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.matchAllQuery());
        searchRequest.source(sourceBuilder);
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits searchHits = response.getHits();
        System.out.println("searchHits.getMaxScore() = " + searchHits.getMaxScore());
        System.out.println("searchHits.getTotalHits() = " + searchHits.getTotalHits());
        SearchHit[] hits = searchHits.getHits();
        for (SearchHit hit : hits) {
            String source = hit.getSourceAsString();
            Goods goods = JSON.parseObject(source, Goods.class);
            System.out.println("goods = " + goods);
        }
    }


    /**
     * 分词查询
     */
    @Test
    public void testFindByTerm() throws IOException {
        SearchRequest searchRequest = new SearchRequest("goods");
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.matchQuery("title", "数码"));
        searchRequest.source(sourceBuilder);
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = response.getHits();
        System.out.println("hits.getTotalHits() = " + hits.getTotalHits());
        System.out.println("hits.getMaxScore() = " + hits.getMaxScore());
        SearchHit[] searchHits = hits.getHits();
        for (SearchHit searchHit : searchHits) {
            String source = searchHit.getSourceAsString();
            Goods goods = JSON.parseObject(source, Goods.class);
            System.out.println("goods = " + goods);
        }
    }

    /**
     * suggest查询
     */
    @Test
    public void testSuggest() throws IOException {
        // 创建SearchRequest，关联索引库
        SearchRequest searchRequest = new SearchRequest("goods");
        // 创建SearchSourceBuilder对象
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        // 创建SuggestBuilder
        SuggestBuilder suggestBuilder = new SuggestBuilder();
        // 调用addSuggestion方法，添加此次查询的名称，以及添加查询的条件
        suggestBuilder.addSuggestion("zdbq",
                // 按照name的前缀为sx自动补全
                SuggestBuilders.completionSuggestion("name").prefix("s").size(10));
        // 把自动补全的条件加到构造工具中
        sourceBuilder.suggest(suggestBuilder);
        // 把构造工具加入到searchRequest对象中
        searchRequest.source(sourceBuilder);
        // 请求
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        // 解析结果
        Suggest.Suggestion<? extends Suggest.Suggestion.Entry<? extends Suggest.Suggestion.Entry.Option>>
                suggestions = response.getSuggest().getSuggestion("zdbq");
        suggestions.forEach(
                suggest ->
                        suggest.getOptions()
                                .forEach(option ->
                                        System.out.println("option.getText() = " + option.getText()))
        );
    }


    /**
     * 异步查询
     */
    @Test
    public void testGetDocumentByIdAsync() throws IOException, InterruptedException {

        System.out.println("准备开始查询");
        // 准备一个查询文档的请求
        GetRequest request = new GetRequest("goods", "1");

        Mono<Goods> goodsMono = Mono.create(monoSink -> {
            // 异步查询一个文档，耗时50ms
            client.getAsync(request, RequestOptions.DEFAULT, new ActionListener<GetResponse>() {
                @Override
                public void onResponse(GetResponse response) {
                    // 获取source
                    String json = response.getSourceAsString();
                    // 把json反序列化
                    Goods goods = JSON.parseObject(json, Goods.class);
                    monoSink.success(goods);

                    System.out.println("查询结束，得到结果： " + goods);
                }

                @Override
                public void onFailure(Exception e) {
                    e.printStackTrace();
                }
            });
        });


        System.out.println("请求已经发出，等待执行结果！");

        Thread.sleep(2000);
    }

    /**
     * 异步删除
     */
    @Test
    public void testAsyncDeleteDocument() throws InterruptedException {
        // 创建请求
        DeleteRequest request = new DeleteRequest("goods", "5");

        // 执行请求，第三个参数是回调处理
        client.deleteAsync(request, RequestOptions.DEFAULT, new ActionListener<DeleteResponse>() {
            /**
             * 执行成功时的回调，参数是IndexResponse结果
             * @param indexResponse 执行结果
             */
            @Override
            public void onResponse(DeleteResponse indexResponse) {
                System.out.println("我是成功的回调！" + indexResponse);
            }
            /**
             * 执行失败时的回调，参数是异常信息
             * @param e 异常信息
             */
            @Override
            public void onFailure(Exception e) {
                System.out.println("我是失败的回调！");
                e.printStackTrace();
            }
        });

        System.out.println("我的异步方法调用完成~~");
        // 因为我们的程序结束会立即停止，接收不到回调结果，这里我们休眠一下，等待下回调结果
        Thread.sleep(2000L);
    }

    /**
     * 异步新增
     */
    @Test
    public void testAsyncAddDocument() throws InterruptedException {
        // 准备文档
        Goods goods = new Goods(5L, "松下电吹风", "松下电吹风 网红电吹风", 1599L);

        // 创建请求
        IndexRequest request = new IndexRequest("goods")
                .id(goods.getId().toString())
                .source(JSON.toJSONString(goods), XContentType.JSON);

        // 执行请求，第三个参数是回调处理
        client.indexAsync(request, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
            /**
             * 执行成功时的回调，参数是IndexResponse结果
             * @param indexResponse 执行结果
             */
            @Override
            public void onResponse(IndexResponse indexResponse) {
                System.out.println("我是成功的回调！" + indexResponse);
            }
            /**
             * 执行失败时的回调，参数是异常信息
             * @param e 异常信息
             */
            @Override
            public void onFailure(Exception e) {
                System.out.println("我是失败的回调！");
                e.printStackTrace();
            }
        });

        System.out.println("我的异步方法调用完成~~");
        // 因为我们的程序结束会立即停止，接收不到回调结果，这里我们休眠一下，等待下回调结果
        Thread.sleep(2000L);
    }

}
